package com.gengzp.mq.server.manager.impl;

import com.gengzp.mq.server.enums.MqMessageStatusEnums;
import com.gengzp.mq.server.facade.MqMessageFacadeService;
import com.gengzp.mq.server.manager.MqResendMessageManager;
import com.gengzp.mq.server.model.message.entity.MqMessage;
import com.gengzp.mq.server.model.thread.MqThreadWaiter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Comparator;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;

import static com.gengzp.mq.server.constants.MqThreadNameConstants.MQ_MESSAGE_RESEND_THREAD_NAME;

/**
 * @ClassName MqResendMessageManagerImpl
 * @Description 消息队列重发消息管理器实现类
 * @Author gengzp
 * @Date 2025/9/4 11:16
 */
@Component
public class MqResendMessageManagerImpl implements MqResendMessageManager {

    private static final Logger logger = LoggerFactory.getLogger(MqResendMessageManagerImpl.class);

    @Autowired
    private MqMessageFacadeService mqMessageFacadeService;

    /**
     * 消息重发线程等待对象
     */
    private static final MqThreadWaiter RESEND_MESSAGE_WAITER = new MqThreadWaiter();

    /**
     * 消息重发队列
     * 由于需要频繁操作排序, 使用 BlockingQueue, 按发送时间升序排序，自带线程安全
     */
    private static final BlockingQueue<MqMessage> RESEND_MESSAGE_QUEUE = new PriorityBlockingQueue<>(
            1024, // 初始容量
            Comparator.comparingLong(MqMessage::getSendTimeStamp)
    );

    @Override
    public void createResendMessageJob() {
        if (RESEND_MESSAGE_WAITER.getThread() != null && RESEND_MESSAGE_WAITER.getThread().isAlive()) {
            logger.error("消息重发线程已存在, 请勿重复创建");
            return;
        }

        RESEND_MESSAGE_WAITER.setThread(new Thread(() -> {
            // 重发消息
            while (true) {
                MqMessage mqMessage = RESEND_MESSAGE_QUEUE.poll();
                if (mqMessage == null) {
                    // 队列为空，等待新消息
                    synchronized (RESEND_MESSAGE_WAITER) {
                        try {
                            RESEND_MESSAGE_WAITER.wait(Long.MAX_VALUE);
                        } catch (Exception e) {
                            logger.error("消息重发线程异常, " + e.getMessage(), e);
                        }
                    }
                } else {
                    // 成功获取到元素, 判断是否已到消息发送时间
                    long delay = mqMessage.getSendTimeStamp() - System.currentTimeMillis();
                    if (delay <= 0) {
                        // 已到发送时间, 立即处理
                        try {
                            mqMessageFacadeService.editMessageById(mqMessage);
                            logger.info("重发消息成功, 消息id: {}, 本次重发次数: {}",
                                    mqMessage.getMessageId(), mqMessage.getRetryCount());
                        } catch (Exception e) {
                            RESEND_MESSAGE_QUEUE.add(mqMessage);
                            logger.info("重发消息失败: {}", e.getMessage());
                        }
                    } else {
                        // 未到重发时间, 继续等待
                        RESEND_MESSAGE_QUEUE.add(mqMessage);
                        synchronized (RESEND_MESSAGE_WAITER) {
                            try {
                                RESEND_MESSAGE_WAITER.wait(delay);
                            } catch (Exception e) {
                                logger.error("消息重发线程异常, " + e.getMessage(), e);
                            }
                        }
                    }
                }
            }
        }, MQ_MESSAGE_RESEND_THREAD_NAME));

        // 执行线程
        RESEND_MESSAGE_WAITER.getThread().start();
        logger.info("消息重发线程 {} 启动成功", MQ_MESSAGE_RESEND_THREAD_NAME);
    }

    @Override
    public void resendMessages(List<MqMessage> mqMessages) {
        // 当前系统时间
        long currentTimeMillis = System.currentTimeMillis();

        // 为消息重新设置 发送时间, 下次重发间隔时间, 重发次数
        for (MqMessage mqMessage : mqMessages) {
            mqMessage.setSendTimeStamp(currentTimeMillis + mqMessage.getNextResendIntervalTime());
            mqMessage.setNextResendIntervalTime(mqMessage.getNextResendIntervalTime() * 2);
            mqMessage.setRetryCount(mqMessage.getRetryCount() + 1);
            mqMessage.setStatus(MqMessageStatusEnums.UNCONSUMED.getCode());
            // 加入重发消息队列
            RESEND_MESSAGE_QUEUE.add(mqMessage);
        }

        // 唤醒重发消息线程
        synchronized (RESEND_MESSAGE_WAITER) {
            RESEND_MESSAGE_WAITER.notify();
        }
    }

}
